package cn.ac.iie.ulss.dataredistribution.handler;

import cn.ac.iie.ulss.dataredistribution.commons.GlobalVariables;
import cn.ac.iie.ulss.dataredistribution.commons.RuntimeEnv;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.MessageExt;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

/**
 *
 * @author: evan
 * @date: 2014-10-15
 */
public class DataAccepter {

    int blockemittersize = 4;
    int mqfetchrunnersize = 12;
    Map<String, ConcurrentLinkedQueue[]> topicToDataPool = null;
    AtomicLong number = new AtomicLong(0);
    String mqnameserver = null;
    ArrayList<DefaultMQPushConsumer> consumers = new ArrayList<DefaultMQPushConsumer>();
    String nodename = null;
    String mqtopicname = null;
    Map<String, BlockEmitter[]> topicToEmitters = null;
    static Logger logger;

    static {
        PropertyConfigurator.configure("log4j.properties");
        logger = org.apache.log4j.Logger.getLogger(DataAccepter.class.getName());
    }

    public void init() {
        blockemittersize = (Integer) RuntimeEnv.getParam(RuntimeEnv.DATA_POOL_COUNT);
        topicToDataPool = (Map<String, ConcurrentLinkedQueue[]>) RuntimeEnv.getParam(GlobalVariables.TOPIC_TO_DATAPOOL);
        mqnameserver = (String) RuntimeEnv.getParam(RuntimeEnv.MQ_NAME_SERVER);
        mqfetchrunnersize = (Integer) RuntimeEnv.getParam(RuntimeEnv.MQ_FETCH_RUNNER_SIZE);
        nodename = (String) RuntimeEnv.getParam(RuntimeEnv.NODENAME);
        mqtopicname = (String) RuntimeEnv.getParam(RuntimeEnv.MQ_TOPIC_NAME);
        topicToEmitters = new HashMap<String, BlockEmitter[]>();
    }

    public void pullDataFromQ() {
        logger.info("begin to pull the data from rmq!");
        String[] topics = mqtopicname.split("\\|");
        for (String topic : topics) {
            initConsumer(topic);
        }
    }

    private void initConsumer(String topic) {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("rd_" + topic);
        consumer.setNamesrvAddr(mqnameserver);
        consumer.setInstanceName("rd_" + topic + "_" + nodename);
        consumer.setConsumeThreadMin(mqfetchrunnersize);
        consumer.setConsumeThreadMax(mqfetchrunnersize);
        consumer.setConsumeConcurrentlyMaxSpan(256);
        consumer.setPullThresholdForQueue(16);
        consumer.setPullBatchSize(16);
        consumer.setConsumeMessageBatchMaxSize(8);
        consumer.setClientCallbackExecutorThreads(32);

        try {
            consumer.subscribe(topic, "*");
            logger.info("rd_" + topic + " consume the topic   " + topic);
        } catch (MQClientException ex) {
            logger.error(ex, ex);
            return;
        }
        ConcurrentLinkedQueue[] dataPool = topicToDataPool.get(topic);
        AtomicLong version = new AtomicLong(1);
        BlockEmitter[] emitters = new BlockEmitter[blockemittersize];
        for (int i = 0; i < blockemittersize; i++) {
            BlockEmitter emitter = new BlockEmitter(dataPool, topic, i, version);
            emitter.init();
            emitters[i] = emitter;
        }
        topicToEmitters.put(topic, emitters);

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                    ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    BlockEmitter[] emitters = topicToEmitters.get(msg.getTopic());
                    Random r = new Random();
                    int v = r.nextInt(emitters.length);
                    emitters[v].emit(msg.getBody());
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        try {
            consumer.start();
            consumers.add(consumer);
        } catch (MQClientException ex) {
            logger.error(ex, ex);
        }
    }

    public void shutdown() {
        for (DefaultMQPushConsumer consumer : consumers) {
            consumer.shutdown();
        }
    }
}
